package com.dyna.cp.kafka.listener;

import com.dyna.constants.TopicConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * @author zhoucaiwang
 * @date 2024/5/17
 */
@Slf4j
@Component
public class KafkaTestListener {

    @KafkaListener(topics = TopicConstant.TEST_TOPIC, groupId = "group1")
    public void testListener(String msg) {
        log.info("\nkafka监听器,接收到消息:{}", msg);
    }

    /**
     * 监听分区0和1的消息
     * 从分区0的起始位置开始消费
     * 从分区2的起始位置开始消费
     * 当给某给分区发送消息超时，说明该分区不存在。
     * 正常情况下给某一个主题发送消息，主题不存在时，会创建一个主题和分区为0，当给0以外的分区发消息时，会因为获取不到该分区而超时。
     */
    @KafkaListener(
            groupId = "partition-group-1",
            containerFactory = "kafkaListenerContainerFactory",
            topicPartitions = {
                    @TopicPartition(
                            topic = TopicConstant.TEST_TOPIC_2,
                            partitionOffsets = {
                                    @PartitionOffset(partition = "0", initialOffset = "0"),
                                    @PartitionOffset(partition = "1", initialOffset = "0")
                            }
                    )
            }
    )
    public void listenPartition01(
            @Payload String message,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {

        log.info("\n<---分区0/1监听器 - 主题: {}, 分区: {}, 内容: {}",
                topic, partition, message);
    }

    /**
     * 监听分区0和2的消息
     * 从分区0的起始位置开始消费
     * 从分区2的起始位置开始消费
     */
    @KafkaListener(
            groupId = "partition-group-1",
            containerFactory = "kafkaListenerContainerFactory",
            topicPartitions = {
                    @TopicPartition(
                            topic = TopicConstant.TEST_TOPIC_2,
                            partitionOffsets = {
                                    @PartitionOffset(partition = "2", initialOffset = "0"),
                                    @PartitionOffset(partition = "3", initialOffset = "0")
                            }
                    )
            }
    )
    public void listenPartition02(
            @Payload String message,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {

        log.info("\n<--- 分区2/3监听器 - 主题: {}, 分区: {}, 内容: {}",
                topic, partition, message);
    }

}
